package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Date:2025/4/9
 * Author：Ys
 * Description:
 *  创建RDD的方式
 *    1. 在程序中创建RDD  从内存中创建RDD
 *    2. 从外部数据源创建RDD
 */
object CreateRDD {



  def main(args: Array[String]): Unit = {
    //1.准备环境                       调用CPU核数  *:全部调用
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    //修改默认分区数
    sparkConf.set("spark.default.parallelism","4")
    //2.让 sc  == SparkContext     spark == Spark SQL
    // 让sc 加载配置 创建Spark Context
    val sc = new SparkContext(sparkConf)

    //1.在内存中创建RDD
    //memoryCreateRDD(sc)

    //2.从外部数据源创建RDD
    fileCreateRDD(sc)
  }
  //在内存中创建RDD
  def memoryCreateRDD(sc:SparkContext): Unit = {
    //从内存中创建RDD，将内存中集合的数据作为处理的数据源
    val rdd1: RDD[Int] = sc.parallelize(1 to 10)
                          //如果只写一个参数 采用的是默认分区。如果没有修改默认分区数和设置RDD分区数，那么默认分区数就是CPU核数
    //RDD[1,2,3,4,5,6,7,9,10]
    val rdd2: RDD[Int] = sc.parallelize(1 to 10,3)//3:分区数
    //获得RDD底层分区数
    println(rdd1.getNumPartitions)
    println(rdd1.partitions.length)

    //将RDD保存到文件中,将计算结果保存到本地系统
    //rdd1.saveAsTextFile("output")
    //rdd2.saveAsTextFile("output1")

    //makeRDD 和 parallelize 是一样，parallelize 是 makeRDD的底层写法
    val rdd3: RDD[Int] = sc.makeRDD(1 to 5)  //常用

    //做RDD的迭代  RDD理解为数组 列表 所以RDD也可以迭代
    //底层原理 先将RDD转换数组，在用数组进行迭代
    rdd3.collect().foreach(println)

    //关闭环境
    sc.stop()
  }


  //在外部文件创建RDD
  def fileCreateRDD(sc: SparkContext): Unit = {
    //1.在项目目录中创建input文件夹，并在该文件夹中创建1.txt文件，在里面写一些东西
    val rdd4: RDD[String] = sc.textFile("input/1.txt")//已经进行分片
    rdd4.collect().foreach(println)

    //关闭环境
    sc.stop()

  }

}
